package top.frsf.rpc.handler;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import top.frsf.rpc.protocol.RpcMessage;
import top.frsf.rpc.protocol.body.RpcRespBody;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RPC 结果响应处理器
 *
 * @author He Kuntao
 * @since 2025/7/8 16:05
 */
@Slf4j
public class RpcResponseHandler extends SimpleChannelInboundHandler<RpcMessage<RpcRespBody>> {

    private static final ConcurrentHashMap<Long, CompletableFuture<RpcRespBody>> pendingRequests = new ConcurrentHashMap<>();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcMessage<RpcRespBody> msg) throws Exception {
        long requestId = msg.getHeader().getRequestId();
        CompletableFuture<RpcRespBody> future = pendingRequests.remove(requestId);
        if (future != null) {
            future.complete(msg.getBody());
        } else {
            log.warn("Received response for unknown request: {}", requestId);
        }
    }

    /**
     * 注册待处理的请求
     *
     * @param requestId 请求ID
     * @return 待处理的请求
     */
    public static CompletableFuture<RpcRespBody> registerPendingRequest(long requestId) {
        CompletableFuture<RpcRespBody> future = new CompletableFuture<>();
        pendingRequests.put(requestId, future);
        return future;
    }
}
